package org.dreamwork.cai.io;

import org.dreamwork.cai.core.BroadcastPayload;
import org.dreamwork.cai.proxy.IServiceResolver;
import org.dreamwork.cai.srf.ServiceRouteFrame;
import org.dreamwork.cai.srf.ServiceRouteTable;
import org.dreamwork.cai.tools.Const;
import org.dreamwork.concurrent.broadcast.ILocalBroadcastReceiver;
import org.dreamwork.concurrent.broadcast.ILocalBroadcastService;
import org.dreamwork.concurrent.broadcast.LocalMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public abstract class AbstractRouteBroadcaster implements IRouteBroadcaster, ILocalBroadcastReceiver {
    private final Logger logger = LoggerFactory.getLogger (AbstractRouteBroadcaster.class);

    protected int port;
    protected String localNamespace;

    protected final IServiceResolver resolver;
    protected final int senderPort;

    protected AbstractRouteBroadcaster (IServiceResolver resolver, int senderPort) {
        this.resolver = resolver;
        this.senderPort = senderPort;

        if (logger.isTraceEnabled ()) {
            logger.trace ("+++++++++++++++++++++++++++");
            logger.trace ("+ sender port = {}", senderPort);
            logger.trace ("+++++++++++++++++++++++++++");
        }
    }

    @Override
    public void update (ServiceRouteFrame frame) throws IOException {
        send (new BroadcastPayload (BroadcastPayload.UPDATE, frame));
    }

    @Override
    public void heartbeat () throws IOException {
        send (new BroadcastPayload (BroadcastPayload.HEARTBEAT));
    }

    @Override
    public void withdrawn () throws IOException {
        send (new BroadcastPayload (BroadcastPayload.WITHDRAWN));
    }

    @Override
    public void bind (int port) throws IOException {
        // 把自己注册为本地广播监听器，用于处理其他节点启动时发送本地 srf
        ILocalBroadcastService service = resolver.get (ILocalBroadcastService.class);
        service.register (Const.Category.CAT_ROUTE_NODE, this);

        bind0 (this.port = port);
    }

    @Override
    public void unbind () throws IOException {
        try {
            withdrawn ();

            unbind0 ();
        } finally {
            // 取消注册的本地广播监听器
            ILocalBroadcastService service = resolver.get (ILocalBroadcastService.class);
            service.unregister (Const.Category.CAT_ROUTE_NODE, this);
        }
    }

    @Override
    public void received (LocalMessage message) {
        if (logger.isTraceEnabled ()) {
            logger.trace ("received a local broadcast: {}", message);
        }
        if (message.what == Const.RouteEvent.STARTUP) {
            ServiceRouteTable table = resolver.get (ServiceRouteTable.class);
            ServiceRouteFrame srf = table.get (localNamespace);
            if (srf == null || srf.items.isEmpty ()) {
                if (logger.isTraceEnabled ()) {
                    logger.trace ("local srf is empty!");
                }
            } else {
                if (logger.isTraceEnabled ()) {
                    logger.trace ("node [{}] startup, send local srf to it.", message.arg);
                }
                BroadcastPayload broadcastPayload = new BroadcastPayload (BroadcastPayload.UPDATE, srf);
                try {
                    send (broadcastPayload);
                } catch (IOException ex) {
                    logger.warn (ex.getMessage (), ex);
                }
            }
        }
    }

    public void setLocalNamespace (String localNamespace) {
        this.localNamespace = localNamespace;
    }

    protected abstract void send (BroadcastPayload payload) throws IOException;

    protected abstract void bind0 (int port) throws IOException;

    protected abstract void unbind0 () throws IOException;
}
